package com.luoqifei;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;

public class MyProcessor  implements Processor<String,String>{
    private ProcessorContext context;
    private KeyValueStore<String,Long> kvStore;
    @Override
    public void init(ProcessorContext processorContext) {
        this.context = processorContext;//keep the processor context locally because we need it in punctuate() and commit
        // call the punctuate every 1000 mill seconds
        this.context.schedule(1000);
        this.kvStore = (KeyValueStore<String, Long>) context.getStateStore("Counts");//get the kv store named 'Counts'
    }

    @Override
    public void process(String s, String line) {
        String[] words = line.toLowerCase().split(" ");
        for (String word: words) {
            Long oldValue = this.kvStore.get(word);
            if(oldValue == null){
                this.kvStore.put(word,1L);
            }else {
                this.kvStore.put(word,oldValue+1);
            }
        }
    }

    @Override
    public void punctuate(long l) {
        KeyValueIterator<String,Long> iter = this.kvStore.all();
        while (iter.hasNext()){
            KeyValue<String,Long> entry = iter.next();
            context.forward(entry.key,entry.value.toString());
        }
        iter.close();
        context.commit();
    }

    @Override
    public void close() {

    }
}
